package com.biconn.bigdata;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.support.SpringBootServletInitializer;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import com.alibaba.fastjson.JSONObject;
import com.biconn.bigdata.es.model.QuestionAndAnswerOnline;
import com.biconn.bigdata.es.repository.BaseElasticsearchRepository;

@SpringBootApplication
@EnableTransactionManagement
@RestController
@EnableAsync
public class StartUpApplication extends SpringBootServletInitializer{
	@Autowired
	private KafkaTemplate< String, String> template ;
	@Autowired
	private CustomCommandLineRunner customCommandLineRunner;
	
    public static void main(String[] args)
    {
        SpringApplication.run(StartUpApplication.class, args);
    	
    }
    
    @Override
    protected SpringApplicationBuilder configure(SpringApplicationBuilder application)
    {
        return application.sources(StartUpApplication.class);
    }
    @Autowired
	private BaseElasticsearchRepository<QuestionAndAnswerOnline> baseElasticsearchRepository;
	
    @GetMapping("api")
	public String getMsg(@RequestParam("key")String key,@RequestParam("content")String content){
		
		Map<String,String> param = new HashMap<>();
		param.put(key, content);
		List<JSONObject>  result = baseElasticsearchRepository.queryOfMatch("question_and_answer_index_stop", "questionAndAnswer", param,0,10);
		return result.toString();
	}
    
    @GetMapping("api/query")
	public String queryMsg(@RequestParam("key")String key,@RequestParam("content")String content){
		
		Map<String,String> param = new HashMap<>();
		param.put(key, content);
		List<Map<String, Object>>  result = baseElasticsearchRepository.queryOfmatchPhrase("question_and_answer_index_test", "questionAndAnswerTest", param,0,10);
		return result.toString();
	}
    
    @GetMapping("api/addField")
	public int addMsg(@RequestParam("id")String id,@RequestParam("key")String key,@RequestParam("content")String content){
		List<Map<String,String>> list = new ArrayList<>();
		Map<String,String> param = new HashMap<>();
		param.put("id", id);
		param.put(key, content);
		list.add(param);
		int  result = baseElasticsearchRepository.batchAddField("question_and_answer_index_test", "questionAndAnswerTest", list);
		return result;
	}
    
    @GetMapping("api/create/index")
	public int addObject(){
    	QuestionAndAnswerOnline questionAndAnswerOnline = new QuestionAndAnswerOnline();
    	baseElasticsearchRepository.saveIndex(questionAndAnswerOnline);
		return 200;
	}
    
    @GetMapping("api/kafka/producer")
	public int producer(){
    	JSONObject it = new JSONObject();
    	it.put("value", "111111111");
    	template.send("question.and.answer.jstorm.bolt.dealed.topic", it.toString());
		return 200;
	}
    
    @GetMapping("api/questionAndAnswer/backup")
	public String questionAndAnswerOntimeDataBackup(){
    	customCommandLineRunner.questionAndAnswerOntimeDataBackup();
		return "questionAndAnswerOntimeDataBackup is working";
	}
    
    @GetMapping("api/deleteDocumentsAndUploadData2ES")
	public String deleteDocumentsAndUploadData2ES(){
    	customCommandLineRunner.deleteDocumentsAndUploadData2ES();
		return "deleteDocumentsAndUploadData2ES is working";
	}
    
    @GetMapping("api/questionAndAnswerTransfer")
	public String questionAndAnswerTransfer(@RequestParam("fromdb")String fromdb,@RequestParam("todb")String todb){
    	customCommandLineRunner.questionAndAnswerOntimeDataTransfer(fromdb,todb);
		return "questionAndAnswerOntimeDataTransfer is working";
	}
    
}
